package com.ice.framework.redission.delaytask;

import com.alibaba.fastjson.JSON;
import com.ice.framework.common.util.MiscUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 *
 * @author wangwei
 * @Date 2022/6/17 16:24
 */
@Slf4j
@Component
public class RedisDelayedQueueInit implements ApplicationContextAware, InitializingBean, DisposableBean {

    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private RedisDelayedQueue redisDelayedQueue;
    @Autowired
    private DelayedQueueConfig delayedQueueConfig;

    private Map<String, RedisDelayedQueueListener> redisDelayedQueueMap = null;
    private boolean threadInterrupt = false;
    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
        this.redisDelayedQueueMap = map;
        this.applicationContext = applicationContext;
    }


    /**
     * 停止
     */
    public void stop() {
        this.threadInterrupt = true;
    }


    @Override
    public void destroy() throws Exception {
        this.stop();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 是否启动延迟任务
        if (delayedQueueConfig.getDelayTaskFlag() && MiscUtil.isNotEmpty(redisDelayedQueue)) {
            ExecutorService executor = null;
            try {
                executor = (ExecutorService) applicationContext.getBean("threadPoolExecutor");
            } catch (BeansException e) {
                executor = (ExecutorService) Executors.newSingleThreadExecutor();
            }
            start(executor);
        }
    }

    /**
     * 每隔1秒遍历队列取值，如果取到值，则延迟消息到期，执行任务
     * @param <T>
     */
    public <T extends DelayTaskBase> void start(ExecutorService executor) {
        Thread delayThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!threadInterrupt) {
                    for (Map.Entry<String, RedisDelayedQueueListener> entry : redisDelayedQueueMap.entrySet()) {
                        RedisDelayedQueueListener queueListener = entry.getValue();
                        String queueName = queueListener.getQueueName();
                        if (MiscUtil.isEmpty(queueName)) {
                            queueName = queueListener.getClass().getName();
                        }
                        RBlockingQueue<T> blockingQueue = redissonClient.getBlockingQueue(queueName);
                        T t = blockingQueue.poll();
                        if (MiscUtil.isNotEmpty(t)) {
                            log.debug("延迟任务队列: {}, 执行内容: {}", queueName, JSON.toJSONString(t));
                            // 执行任务
                            executor.execute(() -> {
                                boolean bool = queueListener.invoke(t);
                                if (!bool) {
                                    if (t.getRetryNum() +1 > t.getRetryMaxNum()){
                                        log.warn("延迟任务未执行成功，超过了重试次数");
                                        queueListener.retryFail(t);
                                    }else{
                                        t.setRetryNum(t.getRetryNum() + 1);
                                        long delay = t.getDelayTime() / (t.getRetryNum()+1);
                                        if (delay < 1){
                                            delay = 1L;
                                        }
                                        redisDelayedQueue.addQueue(t, delay, TimeUnit.SECONDS, t.getQueueName());
                                    }
                                }
                            });
                        }
                    }
                    MiscUtil.sleep(1000);
                }
            }
        });
        delayThread.setName("delay_task_thread");
        delayThread.start();
    }
}
